package com.cet.kafkaproducer.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: cet-practice
 * @description: kafka生产者前置拦截器
 * @author: 陈恩涛
 * @create: 2023-07-26 11:24
 **/

@Slf4j
public class ProducerInterceptorPrefix02 implements ProducerInterceptor<String, String> {

    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger errorCount = new AtomicInteger(0);

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "prefix02_" + record.value();
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue, record.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            successCount.incrementAndGet();
        } else {
            errorCount.incrementAndGet();
        }
    }

    @Override
    public void close() {
        double successRate = successCount.get() * 1.0 / (successCount.get() + errorCount.get());
        System.out.println("interceptor02_successRate: " + successRate);
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
